package net.flyinggroup.vinsguru;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Flux;
import reactor.util.function.Tuples;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.function.Consumer;

@Slf4j
public class SequenceCreator {
    public Consumer<List<Integer>> consumer;

    public Flux<Integer> generateFibonacciWithTuples() {
        return Flux.generate(
                () -> Tuples.of(0, 1),
                (state, sink) -> {
                    sink.next(state.getT1());
                    return Tuples.of(state.getT2(), state.getT1() + state.getT2());
                }
        );
    }

    public Flux<Integer> createNumberSequence() {
        return Flux.create(sink -> {
            log.info("Setup consumer");
            SequenceCreator.this.consumer = items -> items.forEach(sink::next);
        });
    }

    @SneakyThrows
    public static void main(String[] args) {
        SequenceCreator generator = new SequenceCreator();
        List<Integer> sequence1 = generator.generateFibonacciWithTuples().take(3).collectList().block();
        List<Integer> sequence2 = generator.generateFibonacciWithTuples().take(6).collectList().block();

        List<Integer> consolidated = new ArrayList<>();
        generator.createNumberSequence().subscribe(i -> log.info("content: {}", i));

        ExecutorService executor = Executors.newFixedThreadPool(2);
        executor.execute(() -> {
            log.info("setup sequence1");
            generator.consumer.accept(sequence1);
        });
        executor.execute(() -> {
            log.info("setup sequence2");
            generator.consumer.accept(sequence2);
        });


        Thread.sleep(3000);
        executor.shutdownNow();

    }
}
